Deavmi's coding shack
Commit cd686a0b47d839c4d426e7bc9f7e8f3cf06d19f5
Parents : 87f7292
Author : Tristan Brice Velloza Kildaire <deavmi@redxen.eu>
Date : 2026-05-04T19:29:31+02:00
stuf
Changes
7 files changed, 350 insertions(+), 51 deletions(-)
Diff
diff --git a/go.mod b/go.mod
index 3b5df25..472c8f0 100644
--- a/go.mod
+++ b/go.mod
@@ -1,11 +1,13 @@
module new.git.deavmi.assigned.network/deavmi/worcesterinit.git
-go 1.25.8
+go 1.25.9
require new.git.deavmi.assigned.network/deavmi/glog.git v0.0.7
+require new.git.deavmi.assigned.network/deavmi/freefaces.git v0.0.3 // indirect
+
require (
- new.git.deavmi.assigned.network/deavmi/go-niknaks.git v0.0.19
+ new.git.deavmi.assigned.network/deavmi/go-niknaks.git v1.0.3
new.git.deavmi.assigned.network/deavmi/go-tuples.git v1.0.0 // indirect
- new.git.deavmi.assigned.network/deavmi/ini.git v1.0.4
+ new.git.deavmi.assigned.network/deavmi/ini.git v1.0.7
)
diff --git a/go.sum b/go.sum
index 389cf93..40a2b82 100644
--- a/go.sum
+++ b/go.sum
@@ -1,10 +1,12 @@
+new.git.deavmi.assigned.network/deavmi/freefaces.git v0.0.3 h1:at2cViOxfTCsC6Dk92vpP4oj0zNIqeVxE1fVmwPtNz4=
+new.git.deavmi.assigned.network/deavmi/freefaces.git v0.0.3/go.mod h1:CdtAMJElVULEp+P8SOU1QcEqxes2ZaRS66rfsy/rHn8=
new.git.deavmi.assigned.network/deavmi/glog.git v0.0.7 h1:XMIj4KIvyExrlZePv5xLu+gED4yJi5uJnLhiZnrAOlc=
new.git.deavmi.assigned.network/deavmi/glog.git v0.0.7/go.mod h1:nwXj8MbNA84Dr00Duqop8SFoMlFscbMQ/v/17D/nKIs=
new.git.deavmi.assigned.network/deavmi/go-assertions.git v1.0.2 h1:jFcwjgDbJdyxeBqrB/epfS22WjJ9Ug6cYNJFB9I4qVw=
new.git.deavmi.assigned.network/deavmi/go-assertions.git v1.0.2/go.mod h1:5SS262r5xb2YqzOYSIOhpZ8cL/tRF4vYl5lvaVL+evU=
-new.git.deavmi.assigned.network/deavmi/go-niknaks.git v0.0.19 h1:adXciDFvp2aDZDdaLq3+PtvTqZh0NOBSSr9flBQ3OBk=
-new.git.deavmi.assigned.network/deavmi/go-niknaks.git v0.0.19/go.mod h1:Dx7sEGSj9Khi1x0z+spzyKZMCIkLXCJEiNtWxgMbXhM=
+new.git.deavmi.assigned.network/deavmi/go-niknaks.git v1.0.3 h1:HesACzWmW/R6KzR9NsEqqyuWUh8lhrSgQlpew9gP0bA=
+new.git.deavmi.assigned.network/deavmi/go-niknaks.git v1.0.3/go.mod h1:tCh4q52GnOGrl/jeVmBRw7sEyCAdG/7GyU9pdUSQyqc=
new.git.deavmi.assigned.network/deavmi/go-tuples.git v1.0.0 h1:/Kf5IvOuB+Eym84dtoUzKr8NmbHd2tg45QGspbLAm7Y=
new.git.deavmi.assigned.network/deavmi/go-tuples.git v1.0.0/go.mod h1:iOjy/c4BGqXWsEXWvrscX06flHJ/NvAxPOsEC/asLRg=
-new.git.deavmi.assigned.network/deavmi/ini.git v1.0.4 h1:7s9gZv3k9Lta1mbLFdhOuDIfp8s3H6ZC/gR4LYKU0s0=
-new.git.deavmi.assigned.network/deavmi/ini.git v1.0.4/go.mod h1:6FU8CEUtZJgp3l4NmU/tsNqjmJ8inic/5SgyV2ntOJU=
+new.git.deavmi.assigned.network/deavmi/ini.git v1.0.7 h1:VHkQkBsuAq/64UMPH0bLgPXzYaWDMA1OTssGAepnGn4=
+new.git.deavmi.assigned.network/deavmi/ini.git v1.0.7/go.mod h1:CLPCOCRjWuGPoz3UvPUMnvmwrnkGW148XoqgWfQNOVM=
diff --git a/logging.go b/logging.go
new file mode 100644
index 0000000..084c006
--- /dev/null
+++ b/logging.go
@@ -0,0 +1,12 @@
+package main
+
+// something that consumes logs
+//
+// it can be opened, written to,
+// flushed and closed
+type LogConsumer interface {
+ Open() error
+ Write([]byte) error
+ Flush() error
+ Close() error
+}
diff --git a/main.go b/main.go
index a27004d..ea9bc23 100644
--- a/main.go
+++ b/main.go
@@ -117,6 +117,8 @@ func configToUnit(cfg *ini.Ini) (Unit, error) {
return types.Zero[Unit](), e
}
+ l.Debug("servBlock: %v", servBlock)
+
var u Unit = New("")
// required fields (name)
@@ -132,6 +134,7 @@ func configToUnit(cfg *ini.Ini) (Unit, error) {
return types.Zero[Unit](), e
}
u.command = command_r.Value()
+ l.Debug("u.command: %v", u.command)
// optional fields (uid and gid)
if uid_r, e := servBlock.IntAt("uid"); e == nil {
diff --git a/manager.go b/manager.go
index 9edd5ff..436abe4 100644
--- a/manager.go
+++ b/manager.go
@@ -1,10 +1,7 @@
package main
import (
- "context"
"fmt"
- "os"
- "os/exec"
"new.git.deavmi.assigned.network/deavmi/glog.git/std/consts"
stdlogger "new.git.deavmi.assigned.network/deavmi/glog.git/std/logger"
@@ -85,6 +82,7 @@ func (m *Manager) tryStop(u Unit) bool {
return false
}
+// tries to stop the provided unit
func (m *Manager) Stop(unitName string) error {
if f_s, f := m.cache[unitName]; !f {
return fmt.Errorf("Could not stop unit '%s' as it does not exist in the cache", unitName)
@@ -94,6 +92,32 @@ func (m *Manager) Stop(unitName string) error {
}
}
+// marks this unit as launched by:
+//
+// 1. adding it to the run queue (as a LoadedUnit)
+// 2. creates a monitor for it (and associates the LU with it)
+// 3. starts the monitor
+func (m *Manager) addToRunQueue(unit Unit) {
+ m.log.Trace("Starting up '%s'...", unit.Name())
+
+ // move ourselves to the running queue
+ m.running[unit.Name()] = NewLoadedUnit(unit)
+ m.log.Trace("Added '%s' to the run queue", unit.Name())
+
+ // TODO: Add call to launch here
+ monitor := NewMonitor(unit)
+ if e := monitor.Start(); e == nil {
+ m.log.Info("Started task %v", unit)
+ } else {
+ // TODO: Return as an error so we can remove ourselves
+ // from run queue and add ourselves to error queue?
+ //
+ // TODO: Actually we should just do that ourseles
+ m.log.Error("Could not start")
+ }
+
+}
+
func (m *Manager) tryStart(unitName string) bool {
if f_s, f := m.cache[unitName]; f {
// are we already loaded?
@@ -120,10 +144,8 @@ func (m *Manager) tryStart(unitName string) bool {
m.log.Trace("Were all %d dependencies satisfied for '%s'?: %v", len(f_deps), unitName, allSatisfied)
if allSatisfied {
- // TODO: Add call to launch here
-
- // move ourselves to the running queue
- m.running[unitName] = NewLoadedUnit(f_s)
+ // try to launch it
+ m.addToRunQueue(f_s)
}
//
@@ -227,40 +249,3 @@ func (m *Manager) canLaunch(dep string) bool {
return false
}
-
-// TODO: Take in a cancellable-context
-func (m *Manager) doProcessLaunch(ctx context.Context, u Unit) {
- // configure a new command to execute
- var e *exec.Cmd = exec.CommandContext(ctx, u.command, u.args...)
-
- // install environment variables
- var envStrs []string = m.makeEnvs(u)
- m.log.Debug("Installed %d environment variables for %s", len(envStrs), u.name)
- e.Env = envStrs
-
- // set working directory (if not specified then set to directory of `command`
- // if it is an absolute or we can find it from `command` or where service file
- // is located)
- // e.Dir
-
- // after this c.Process becomes non-nil and we should
- // release resources (when process is done, via e.Wait()?)
- e.Start()
- var proc *os.Process = e.Process
- m.log.Debug("Process PID %d for '%s'...", proc.Pid, u.Name())
-
-}
-
-func (m *Manager) makeEnvs(u Unit) []string {
- var vars []string
- for k, v := range u.GetEnvs() {
- varStr := makeEnvString(k, v)
- vars = append(vars, varStr)
- m.log.Trace("Will install environment variable %s into %s", u.name, varStr)
- }
- return vars
-}
-
-func makeEnvString(name string, value string) string {
- return fmt.Sprintf("%s=%s", name, value)
-}
diff --git a/manager_test.go b/manager_test.go
index ff8fae2..ee9056e 100644
--- a/manager_test.go
+++ b/manager_test.go
@@ -8,17 +8,20 @@ func Test_deps(t *testing.T) {
// create dependency `a` which depends on `b`
// and load it
a := New("a")
+ a.command = "/bin/yes"
a.AddDep("b")
m.Load(a)
// create dependency `b` which depends on `c`
// and load it
b := New("b")
+ b.command = "/bin/yes"
b.AddDep("c")
m.Load(b)
// create dependency `c` and load it
c := New("c")
+ c.command = "/bin/yes"
m.Load(c)
// m.canLaunch()
diff --git a/monitor.go b/monitor.go
new file mode 100644
index 0000000..d5cb9c1
--- /dev/null
+++ b/monitor.go
@@ -0,0 +1,292 @@
+package main
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "os"
+ "os/exec"
+ "sync"
+
+ "new.git.deavmi.assigned.network/deavmi/glog.git/std/consts"
+ stdlogger "new.git.deavmi.assigned.network/deavmi/glog.git/std/logger"
+ "new.git.deavmi.assigned.network/deavmi/go-niknaks.git/slices"
+)
+
+func (b Bruh) Name() string {
+ return b.u.Name()
+}
+
+type Bruh struct {
+ // original unit
+ u Unit
+
+ // pointer to the loaded unit
+ lu *LoadedUnit
+
+ // pointer to the manager instance
+ m *Manager
+
+ // the command
+ cmd *exec.Cmd
+
+ // pipes monitored
+ pipes []*PipeMonitor
+ pipesLock sync.Mutex
+ // pipes []io.Closer
+
+ // control signaling
+ ctx context.Context
+ cancelFunc context.CancelCauseFunc
+
+ // logging
+ log *stdlogger.StdLogger
+}
+
+func NewMonitor(unit Unit) Bruh {
+ return NewMonitorWithCtx(unit, context.Background())
+}
+
+// creates a new monitor using the details
+// from the provided Unit and inherits the
+// given Context
+func NewMonitorWithCtx(unit Unit, ctx context.Context) Bruh {
+ var b Bruh = Bruh{u: unit}
+
+ // derive from the context one that we can get
+ // a cancelation function
+ //
+ // this function let's us set a cause and then
+ // cancel
+ b.ctx, b.cancelFunc = context.WithCancelCause(ctx)
+
+ // Setup logging
+ // TODO: Later inherit a logger passed in
+ b.log = stdlogger.NewStdLogger(fmt.Sprintf("Monitor %s", unit.Name()))
+ b.log.SetLevel(consts.LEVEL_TRACE)
+
+ return b
+}
+
+func (b *Bruh) Start() error {
+ // configure a new command to execute and use
+ // the context passed in
+ //
+ // this means when WE receive a cancel we can
+ // get it but so will all derived contexts
+ var e *exec.Cmd = exec.CommandContext(b.ctx, b.u.command, b.u.args...)
+ b.log.Trace("b.u.command: %v", b.u.command)
+
+ // install environment variables
+ var envStrs []string = b.makeEnvs(b.u)
+ b.log.Debug("Installed %d environment variables for %s", len(envStrs), b.u.name)
+ e.Env = envStrs
+
+ // set working directory (if not specified then set to directory of `command`
+ // if it is an absolute or we can find it from `command` or where service file
+ // is located)
+ // e.Dir
+
+ // set the standard output and standard error writers
+ // (that the process should write to)
+
+ // after this c.Process becomes non-nil and we should
+ // release resources (when process is done, via e.Wait()?)
+ if s_e := e.Start(); s_e != nil {
+ b.log.Error("Error starting process for '%s': %v", b.u.Name(), s_e)
+ return fmt.Errorf("Error starting process for '%s': %v", b.u.Name(), s_e)
+ }
+ var proc *os.Process = e.Process
+ b.log.Debug("Process PID %d for '%s'...", proc.Pid, b.u.Name())
+
+ // schedule a goroutine here with the monitor
+ // function
+ go b.monitor()
+
+ // TODO: We also want to start some go routine that can
+ // read from the standard out and collect logs for us
+ std_out, err := e.StdoutPipe()
+ if err != nil {
+ // TODO: Handle this
+ b.log.Error("Could not obtain standard output pipe for '%s': %v", b.u.Name(), err)
+ return fmt.Errorf("Could not obtain standard output pipe for '%s': %v", b.u.Name(), err)
+ } else {
+ // create a new pipe minitor and start it up
+ NewPipeMonitor(b, std_out, b.ctx).Start()
+ }
+
+ std_err, err := e.StderrPipe()
+ if err != nil {
+ // TODO: Handle this
+ b.log.Error("Could not obtain standard error pipe for '%s': %v", b.u.Name(), err)
+ return fmt.Errorf("Could not obtain standard error pipe for '%s': %v", b.u.Name(), err)
+ } else {
+ // create a new pipe minitor and start it up
+ NewPipeMonitor(b, std_err, b.ctx).Start()
+ }
+
+ return nil
+}
+
+func (b *Bruh) AddPipeMon(pipeMon *PipeMonitor) {
+ b.pipesLock.Lock()
+ b.pipes = append(b.pipes, pipeMon)
+ b.pipesLock.Unlock()
+}
+
+func (b *Bruh) RemovePipeMon(pipeMon *PipeMonitor) {
+ b.pipesLock.Lock()
+ b.pipes = slices.RemoveElem(pipeMon, b.pipes)
+ b.pipesLock.Unlock()
+}
+
+func NewPipeMonitor(monitor *Bruh, r io.ReadCloser, ctx context.Context) PipeMonitor {
+ return PipeMonitor{m: monitor, pipe: r, ctx: ctx}
+}
+
+func (b *Bruh) monitor() {
+ // wait for us to be cancelled
+ select {
+ case <-b.ctx.Done():
+ b.log.Info("Got closing signal: %v", context.Cause(b.ctx))
+ }
+
+ // TODO: Perform shutdown
+ b.log.Trace("Shutting down pipes...")
+ for p_idx, p := range b.pipes {
+ b.log.Trace("(%d) Shutting down pipe '%v'...", p_idx, p)
+ p.Stop()
+ }
+}
+
+const BUFFER_SIZE = 1000
+
+// combines the monitored pipe
+// with the destination of the
+// read data
+type PipeMonitor struct {
+ // the Monitor we are attached to
+ m *Bruh
+
+ // pipe
+ pipe io.ReadCloser
+
+ // context
+ ctx context.Context
+
+ // logging
+ log *stdlogger.StdLogger
+
+ // consumer of logs (optional)
+ consumer LogConsumer
+}
+
+func (pm PipeMonitor) String() string {
+ return fmt.Sprintf("%T (unit: %s, pipe: %v)", pm, pm.m.Name(), pm.pipe)
+}
+
+// Spawns a go routine to start this PipeMonitor
+//
+// This will eventually cause it to be added to
+// the Monitor's pipe list
+func (pm PipeMonitor) Start() {
+ go pm.logMonitor()
+}
+
+// Stops the pipe monmitor
+//
+// This is accomplished by:
+//
+// 1. Updating the context (TODO: not needed but it is nice)
+// 2. Close()'ing the ReadCloser calling the PipeMonitor's
+// `Read()` to unblock and the Goroutine to end
+func (pm PipeMonitor) Stop() {
+ pm.log.Debug("Stopping %v...", pm)
+ if e := pm.pipe.Close(); e != nil {
+ pm.log.Error("Error when closing internal pipe during shutdown: %v", e)
+ }
+ pm.log.Debug("%v stopped", pm)
+}
+
+// log monitor for the given reader
+//
+// waits for bytes and when it gets
+// an error then it exits.
+func (b *PipeMonitor) logMonitor() {
+ b.log.Trace("Log monitor for '%s' on fd '%v' enter", b.m.Name(), b.pipe)
+
+ // Add `r` to b.pipes
+ b.m.AddPipeMon(b)
+
+ if b.consumer != nil {
+ // if it fails to open then nil it out so we don't
+ // try and use it later
+ if e := b.consumer.Open(); e != nil {
+ b.log.Error("Could not open LogConsumer, ignoring it...: %v", e)
+ b.consumer = nil
+ }
+ }
+
+ b.log.Trace("Log monitor for '%s' on fd '%v' using buffer size %d", BUFFER_SIZE)
+
+ var buff []byte = make([]byte, BUFFER_SIZE)
+ for {
+ // we read at most `BUFFER_SIZE` bytes per call
+ n, e := b.pipe.Read(buff)
+ if e != nil {
+ // TODO: Check context in case maybe something basd happened
+ // like process closed one end of pipe _instead_ of us initiating
+ // a tare down
+ b.log.Debug("Log monitor error returned whilst reading: %v", e)
+ break
+ }
+ buff = buff[0:n] // trim to bytes got
+ b.log.Trace("Got %d bytes: '%s'", n, (string)(buff))
+
+ // If a consumer is attached then push the
+ // bytes to it
+ if b.consumer != nil {
+ if e := b.consumer.Write(buff); e != nil {
+ b.log.Error("Error writing %d bytes to consumer %v: %v", n, b.consumer, e)
+ }
+ }
+
+ // TODO: When to flush consumer?
+ }
+
+ b.log.Trace("Log monitor for '%s' on fd '%v' exit", b.m.Name(), b.pipe)
+
+ // Remove `r` from b.pipes
+ b.m.RemovePipeMon(b)
+
+ // Shutdown the LogConsumer
+ if b.consumer != nil {
+ b.consumer.Flush()
+ b.consumer.Close()
+ }
+}
+
+// signal
+// Stop the process
+//
+// 1. calls `Kill()`
+// 2. cancels the `b.ctx`
+// 2.1 this unblocks Goroutine running `monitor()`
+// 2.2 this causes the process in `b.cmd.process` to kill the process
+func (b *Bruh) Stop() {
+
+}
+
+func (m *Bruh) makeEnvs(u Unit) []string {
+ var vars []string
+ for k, v := range u.GetEnvs() {
+ varStr := makeEnvString(k, v)
+ vars = append(vars, varStr)
+ m.log.Trace("Will install environment variable %s into %s", u.name, varStr)
+ }
+ return vars
+}
+
+func makeEnvString(name string, value string) string {
+ return fmt.Sprintf("%s=%s", name, value)
+}
Served by rngit 1.5.0 - Generated in 0.15s